package com.chukun.flink.table.api.sql;

import com.chukun.flink.table.bean.ClickBean;
import com.chukun.flink.table.source.PrepareData;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.sql.Timestamp;

/**
 * @author chukun
 * @version 1.0.0
 * @description sql分组基本使用
 * @createTime 2022年06月02日 23:36:00
 */
public class GroupTemplate {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        DataStreamSource<ClickBean> dataStream = env.fromCollection(PrepareData.getClicksData());

        // 定义输入表的数据格式列
        Table inputTable = tableEnv.fromDataStream(dataStream, Schema.newBuilder()
                .column("id", DataTypes.INT())
                .column("user", DataTypes.STRING())
                .column("url", DataTypes.STRING())
                .column("time", DataTypes.TIMESTAMP().bridgedTo(Timestamp.class))
                .build());

        tableEnv.createTemporaryView("Clicks", inputTable);

        // 查询sql语句
        Table result = tableEnv.sqlQuery("select user as name, count(url) as number from Clicks group by user");

        DataStream<Row> rowDataStream = tableEnv.toChangelogStream(result);

        rowDataStream.print("查询结果");

        env.execute("GroupTemplate");
    }
}
